package org.jivesoftware.openfire.plugin.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.jivesoftware.openfire.plugin.chatlogs.DbChatLogsManager;
import org.jivesoftware.util.JiveGlobals;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import static org.apache.kafka.common.utils.Utils.getKafkaClassLoader;

public class ChatLogsConsumer implements Runnable {

    private static KafkaConsumer<String, String> consumer;

    private static final Logger log = LoggerFactory.getLogger(ChatLogsConsumer.class);

    static {
        log.info("@@@@@@@@ kafka 消费者初始化开始");
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, JiveGlobals.getProperty(KafkaUtil.KAFKA_BOOTSTRAP_SERVERS));
        props.put(ConsumerConfig.GROUP_ID_CONFIG, JiveGlobals.getProperty(KafkaUtil.KAFKA_GROUP_ID));
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, JiveGlobals.getIntProperty(KafkaUtil.KAFKA_MAX_POLL_RECORDS, 100));
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, JiveGlobals.getIntProperty(KafkaUtil.KAFKA_SESSION_TIMEOUT_MS, 25000));
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        //server发送到消费端的最小数据，若是不满足这个数值则会等待直到满足指定大小。默认为1表示立即接收
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1);
        //若是不满足fetch.min.bytes时，等待消费端请求的最长等待时间
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 1000);
        try {
            consumer = new KafkaConsumer<>(props);
            log.info("@@@@@@@@ kafka 消费者初始化成功");
        } catch (Exception e) {
            log.info("@@@@@@@@ kafka 消费者初始化失败");
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        // 设置消费组订阅的topic (topic需要先在控制台进行创建)
        List<String> subscribedTopics = new ArrayList<String>();
        subscribedTopics.add(KafkaUtil.KAFKA_TOPIC_VALUE);
        consumer.subscribe(subscribedTopics);
        while (true) {
            try {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                for (ConsumerRecord<String, String> record : records) {
                    DbChatLogsManager.getInstance().handle(record.value());
                }
            } catch (Exception e) {
                try {
                    Thread.sleep(1000);
                } catch (Throwable i) {
                }

                log.error("consumer 处理错误 : ", e);
            }
        }
    }
}
